package cn.doit.commons

import cn.doit.bean.LogBean
import cn.doit.utils.JedisUtils
import org.apache.spark.rdd.RDD
import org.slf4j.LoggerFactory

/**
  * Created by Administrator on 2018/11/16.
  */
object UVAnalys {
  val logger = LoggerFactory.getLogger(this.getClass.getSimpleName)

  /**
    * 实时访问 uv
    *
    * @param beanRdd
    */
  def realFlowUV(beanRdd: RDD[LogBean]) = {
    //去重数据
    val disData = beanRdd.map(fe => {
      val day = fe.getDay
      val hour = fe.getHour
      val minute = fe.getMinute
      val siteId = fe.getSiteId
      val uId = fe.getUid
      (day, hour, minute, siteId, uId)
    }).distinct()

    disData.foreachPartition(fp => {
      lazy val jedis = JedisUtils.getJedis()

      try {
        fp.foreach {
          case (day, hour, minute, siteId, uId) => {
            val uvRedisKey = "FLOWREAL@UV@" + siteId + "@" + day + "@" + hour + "@" + minute
            jedis.hset(uvRedisKey, uId, "1")

            if (!"1".equals(siteId)) {
              val uvRedisKey_tmp = "FLOWREAL@UV@1" + "@" + day + "@" + hour + "@" + minute
              jedis.hset(uvRedisKey_tmp, uId, "1")
            }
            //jedis.scard(uvRedisKey)//返回key的value数量
            // 设置key有效期
            // jedis.expire(uvRedisKey, 24 * 60 * 60)
          }
        }
      } catch {
        case e: Exception => {
          logger.error("UVAnalys.realFlowUV foreach error !!!!!!!!!!!!!!!! ")
          e.printStackTrace()
        }
      } finally {
        jedis.close()
      }
    })
  }

}
